package com.alibaba.otter.canal.client.adapter.rdb.support;

import com.alibaba.otter.canal.client.adapter.rdb.config.RDBConstants;
import com.alibaba.otter.canal.client.adapter.support.Util;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.io.Closeable;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * sql批量执行器
 *
 * @author rewerma 2018-11-7 下午06:45:49
 * @version 1.0.0
 */
public class BatchExecutor implements Closeable {

    private static final Logger logger = LoggerFactory.getLogger(BatchExecutor.class);

    private DataSource          dataSource;
    private Connection          conn;
    private AtomicInteger       idx    = new AtomicInteger(0);

    public BatchExecutor(DataSource dataSource){
        this.dataSource = dataSource;
    }

    public Connection getConn() {
        try {
            if (conn == null || conn.isClosed()) {
                try {
                    conn = dataSource.getConnection();
                    this.conn.setAutoCommit(false);
                } catch (SQLException e) {
                    logger.error(e.getMessage(), e);
                }
            }
        } catch (SQLException e) {
            logger.error("获取数据库连接异常: {}", e.getMessage());
        }
        return conn;
    }

    public static void setValue(List<Map<String, ?>> values, int type, Object value) {
        Map<String, Object> valueItem = new HashMap<>();
        valueItem.put("type", type);
        valueItem.put("value", value);
        values.add(valueItem);
    }

    public static void setValue(List<Map<String, ?>> values, String name, int type, Object value, Map<String, String> keywordsIdentifier) {
        Map<String, Object> valueItem = new HashMap<>();
        valueItem.put("name", name);
        valueItem.put("type", type);
        valueItem.put("value", value);
        valueItem.put(RDBConstants.KEYWORDS_IDENTIFIER_PREFIX, keywordsIdentifier.get(RDBConstants.KEYWORDS_IDENTIFIER_PREFIX));
        valueItem.put(RDBConstants.KEYWORDS_IDENTIFIER_SUFFIX, keywordsIdentifier.get(RDBConstants.KEYWORDS_IDENTIFIER_SUFFIX));
        values.add(valueItem);
    }

    public void execute(String sql, List<Map<String, ?>> values) throws SQLException {
        PreparedStatement pstmt = getConn().prepareStatement(sql);
        int len = values.size();
        for (int i = 0; i < len; i++) {
            int type = (Integer) values.get(i).get("type");
            Object value = values.get(i).get("value");
            SyncUtil.setPStmt(type, pstmt, value, i + 1);
        }

        pstmt.execute();
        idx.incrementAndGet();
        pstmt.close();
    }

    // 如果是先 insert 再delete 再insert的情况，会导致数据丢失，因此需要特殊处理
    public void checkDelete(Connection connection, String insertSql, Map<String, List<List<Map<String, ?>>>> deleteSqlValuesMap) {
        if (null != deleteSqlValuesMap && deleteSqlValuesMap.size() > 0) {
            // 获取insert语句中的表名
            String tableName = insertSql.substring(insertSql.indexOf("INTO") + 4, insertSql.indexOf("(")).trim();
            // 组装字段名和字段值为map，方便使用
            String[] fieldNames = insertSql.substring(insertSql.indexOf("(") + 1, insertSql.indexOf(") VALUES")).split(",");
            String[] fieldValues = insertSql.substring(insertSql.indexOf(") VALUES (") + 10, insertSql.lastIndexOf(")")).split(",");
            Map<String, String> insertMap = new ConcurrentHashMap<>();
            for (int i = 0, len = fieldNames.length; i < len; i++) {
                if (fieldValues[i].startsWith("'") && fieldValues[i].endsWith("'")) {
                    insertMap.put(fieldNames[i], fieldValues[i].substring(1, fieldValues[i].length() - 1));
                } else {
                    insertMap.put(fieldNames[i], fieldValues[i]);
                }
            }

            List<String> deleteSqls = new ArrayList<>(deleteSqlValuesMap.keySet());
            // 遍历所有delete语句
            for (String deleteSql : deleteSqls) {
                // 有该表的delete语句时
                if (deleteSql.contains("DELETE FROM " + tableName)) {
                    // 该表所有删除语句的值
                    List<List<Map<String, ?>>> delValues = deleteSqlValuesMap.get(deleteSql);
                    String prefix = "";
                    String suffix = "";
                    final Map<String, ?> tmp_value = delValues.get(0).get(0);
                    if (tmp_value.containsKey(RDBConstants.KEYWORDS_IDENTIFIER_PREFIX) && tmp_value.containsKey(RDBConstants.KEYWORDS_IDENTIFIER_SUFFIX)) {
                        prefix = tmp_value.get(RDBConstants.KEYWORDS_IDENTIFIER_PREFIX).toString();
                        suffix = tmp_value.get(RDBConstants.KEYWORDS_IDENTIFIER_SUFFIX).toString();
                    }

                    for (int index = 0, size = delValues.size(); index < size; index++) {
                        List<Map<String, ?>> values = delValues.get(index);
                        boolean flag = true;
                        // 单条删除语句中每个字段的值，即所有主键的字段名和字段值
                        for (Map<String, ?> value : values) {
                            // 获取insert语句中该字段的值
                            String val = insertMap.get(prefix + value.get("name").toString() + suffix);
                            if (!StringUtils.isEmpty(val)) {
                                // 如果insert语句中主键的值和delete语句中主键的值相等
                                if (!val.equals(value.get("value").toString())) {
                                    flag = false;
                                }
                            } else {
                                flag = false;
                            }
                        }
                        // 如果insert语句中所有主键的值和delete语句中主键的值均相等,则说明此时的主键冲突是因为出现了先delete再insert的情况导致的（但是代码逻辑统一先insert再delete，造成数据丢失）
                        if (flag) {
                            Savepoint savepoint = null;
                            try (
                                    // 删除记录
                                    PreparedStatement pstmt = connection.prepareStatement(deleteSql);
                                    // 执行insert
                                    Statement statement = connection.createStatement()
                            ) {
                                savepoint = connection.setSavepoint();

                                int len = values.size();
                                for (int i = 0; i < len; i++) {
                                    int type = (Integer) values.get(i).get("type");
                                    Object value = values.get(i).get("value");
                                    SyncUtil.setPStmt(type, pstmt, value, i + 1);
                                }
                                int execute = pstmt.executeUpdate();
                                if (execute != 1) {
                                    logger.error("删除主键冲突记录失败,{}", execute);
                                }

                                int executeUpdate = statement.executeUpdate(insertSql);
                                if (executeUpdate != 1) {
                                    logger.error("删除后执行插入失败，{}", executeUpdate);
                                }

                                // 移除已执行的delete语句
                                List<List<Map<String, ?>>> lists = deleteSqlValuesMap.get(deleteSql);
                                List<List<Map<String, ?>>> lists_copy = new ArrayList<>();
                                for (int i = 0, length = lists.size(); i < length; i++) {
                                    if (i != index) {
                                        lists_copy.add(lists.get(i));
                                    }
                                }
                                deleteSqlValuesMap.put(deleteSql, lists_copy);
                            } catch (SQLException e) {
                                try {
                                    if (null != savepoint) {
                                        connection.rollback(savepoint);
                                    }
                                } catch (SQLException e1) {
                                    logger.error("检测到主键冲突，尝试解决时发生异常，尝试将所做操作回滚时异常,{}", e.toString());
                                }
                                logger.error("尝试解决主键冲突时发生异常,{}", e.toString());
                            }
                            // 当一个批量sql中对同一条数据进行多次增删时，一次只处理一次增删操作，否则将导致最终数据为第一次发生主键冲突时的值
                            // 例： 新增id:6,age:10,然后删除，再新增id:6,age:20,然后再删除，再新增id:6,age:30
                            // 如果不加break将导致最终数据为id:6,age:20（正确的数据应为id:6,age:30）
                            return;
                        }
                    }

                }
            }


        }
    }


    // 自己组装成批量SQL后使用Statement将所有表的SQL都加入到批中一起发送给数据库（考虑不同的字段数据类型带来的问题）
    public void batchSqlExecute(Map<String, List<List<Map<String, ?>>>> insertSqlValuesMap, Map<String, List<List<Map<String, ?>>>> deleteSqlValuesMap) throws SQLException {
        Connection conn = getConn();
        conn.setAutoCommit(false);

        Statement statement = conn.createStatement();

        String databaseProductName = getDatabaseProductName();
        if ("postgresql".equalsIgnoreCase(databaseProductName)) {
            // 关闭查询优化器，防止出现 Error "Multiple updates to a row by the same query is not allowed"
            statement.executeUpdate("set optimizer=off");
        }

        List<String> sqls = new ArrayList<>();

        int count = insertSqlValuesMap.size();

        for (Map.Entry<String, List<List<Map<String, ?>>>> kv : insertSqlValuesMap.entrySet()) {
            String sql = compose2BatchSql(kv.getKey(), kv.getValue());
            count--;
            if (!StringUtils.isEmpty(sql)) {
                sqls.add(sql);
                statement.addBatch(sql);
                if (sqls.size() % 500 == 0 || count == 0) {
                    try {
                        statement.executeBatch();
                    } catch (SQLException e) {
                        // 回滚
                        conn.rollback();
                        // postgresql 主键冲突错误信息
                        if (e.getMessage().contains("duplicate key") || e.getMessage().contains("Duplicate entry") || e.getMessage().startsWith("ORA-00001:")) {
                            // TODO 增加更多关系数据库的主键冲突的错误码
                            Statement statement_try = conn.createStatement();
                            Savepoint savepoint=null;
                            for (String sql_ : sqls) {
                                // 此处sql为组装sql，需拆分为单条sql
                                String insertSqlPre = sql_.substring(0, sql_.indexOf("VALUES") + 6);
                                List<String> values = Arrays.asList(sql_.substring(sql_.indexOf("VALUES (") + 8).split("\\),\\("));
                                // 去除最后一个value的右括号
                                int last = values.size() - 1;
                                String v = values.get(last);
                                values.set(last, v.substring(0, v.length() - 1));
                                for (String value : values) {
                                    String execSql = insertSqlPre + " (" + value + ")";
                                    try {
                                        if ("postgresql".equalsIgnoreCase(conn.getMetaData().getDatabaseProductName())) {
                                            savepoint = conn.setSavepoint();
                                        }
                                        statement_try.executeUpdate(execSql);
                                    } catch (SQLException e1) {
                                        if (e1.getMessage().contains("duplicate key")) {
                                            if (null != savepoint) {
                                                conn.rollback(savepoint);
                                            }
                                            // 如果是先 insert 再delete 再insert的情况，会导致数据丢失，因此特殊处理
                                            // checkDelete(conn, execSql, deleteSqlValuesMap);
                                        } else if (e1.getMessage().contains("Duplicate entry") || e1.getMessage().startsWith("ORA-00001:")) {
                                            // ignore
                                            // TODO 增加更多关系数据库的主键冲突的错误码
                                            // 如果是先 insert 再delete 再insert的情况，会导致数据丢失，因此特殊处理
                                            // checkDelete(conn, execSql, deleteSqlValuesMap);
                                        } else {
                                            logger.error("单条SQL执行错误信息：{}", e1.toString());
                                            throw e1;
                                        }
                                    }
                                }
                            }

                            statement_try.close();
                            // 清除已执行的SQL
                            sqls.clear();
                        } else {
                            logger.error("批量SQL执行错误信息：{}", e.toString());
                            throw e;
                        }
                    }
                }
            } else {
                if (logger.isDebugEnabled()) {
                    logger.debug("###### 执行时跳过: {} ######", kv.getKey());
                }
            }
        }
        conn.commit();
        statement.close();
    }

    // 组装批量sql
    public String compose2BatchSql(String sql, List<List<Map<String, ?>>> values) {
        // 获取数据库产品名
        String dbProductName = getDatabaseProductName();

        StringBuilder composeSql = new StringBuilder();
        // 当前Sql对应的批量数据数
        int dataSize = values.size();
        // 仅有一条数据，无需组装，只需将占位符替换为对应的值即可
        if (dataSize == 1) {
            // 防止数据中含有 ?
            composeSql.append(sql.replace("?", "\001"));
            // 替换 ?
            for (Map<String, ?> val : values.get(0)) {
                int index = composeSql.indexOf("\001");
                composeSql.deleteCharAt(index);
                composeSql.insert(index, formatValue((Integer) (val.get("type")), val.get("value")));
            }
            return composeSql.toString();
        } else if (dataSize > 1) {
            // 多条数据则组装
            // 获取关键字标识符
            String prefix = "";
            String suffix = "";
            final Map<String, ?> tmp_value = values.get(0).get(0);
            if (tmp_value.containsKey(RDBConstants.KEYWORDS_IDENTIFIER_PREFIX) && tmp_value.containsKey(RDBConstants.KEYWORDS_IDENTIFIER_SUFFIX)) {
                prefix = tmp_value.get(RDBConstants.KEYWORDS_IDENTIFIER_PREFIX).toString();
                suffix = tmp_value.get(RDBConstants.KEYWORDS_IDENTIFIER_SUFFIX).toString();
            }

            // 组装insert语句
            if (sql.startsWith("INSERT")) {
                composeSql.append(sql, 0, sql.indexOf("VALUES") + 7);

                for (List<Map<String, ?>> value : values) {
                    composeSql.append("(");
                    for (Map<String, ?> val : value) {
                        composeSql.append(formatValue((Integer) (val.get("type")), val.get("value"))).append(",");
                    }
                    composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                    composeSql.append("),");
                }

                composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                return composeSql.toString();

            } else if (sql.startsWith("UPDATE")) {
                // UPDATE test SET name=?, name2=? WHERE id=? AND id2=?
                // 组装update语句成 update test set name = tmp.name from (values (1,'zhangsan'),(2,'lisi'),(3,'wangwu')) as tmp(id,name) where test.id=tmp.id;
                if ("postgresql".equalsIgnoreCase(dbProductName)) {
                    // 获取主键
                    List<String> keys = Arrays.asList(sql.substring(sql.indexOf("WHERE") + 6).replace("=?", "").split(" AND "));
                    // 修改的字段
                    List<String> fields = Arrays.asList(sql.substring(sql.indexOf("SET") + 4, sql.indexOf("WHERE") - 1).replace("=?", "").split(", "));

                    List<String> all = new ArrayList<>(keys);
                    all.addAll(fields);

                    composeSql.append(sql, 0, sql.indexOf("SET")).append("t SET ");
                    for (String field : fields) {
                        composeSql.append(field).append("=tmp.").append(field).append(",");
                    }
                    composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                    composeSql.append(" FROM (VALUES ");

                    for (List<Map<String, ?>> value : values) {
                        composeSql.append("(");

                        for (String field : all) {
                            for (Map<String, ?> val : value) {
                                if (field.replace(prefix, "").replace(suffix, "").trim().equals(val.get("name").toString())) {
                                    composeSql.append(formatValue((Integer) (val.get("type")), val.get("value"))).append(",");
                                    break;
                                }
                            }
                        }

                        composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                        composeSql.append("),");
                    }
                    composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                    composeSql.append(") AS tmp(");
                    for (String field : all) {
                        composeSql.append(field).append(",");
                    }
                    composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                    composeSql.append(") WHERE ");
                    for (String key : keys) {
                        composeSql.append("t.").append(key).append("=tmp.").append(key).append(" AND ");
                    }
                    composeSql.delete(composeSql.lastIndexOf(" AND "), composeSql.length());

                    String lastSql = composeSql.toString();

                    List<Map<String, ?>> maps = values.get(0);

                    // 添加日期类型转换
                    for (String field : all) {
                        for (Map<String, ?> val : maps) {
                            String name = prefix + val.get("name").toString() + suffix;
                            if (field.trim().equals(name)) {
                                lastSql = insertTypeCast(lastSql, name, (Integer) (val.get("type")));
                                break;
                            }
                        }
                    }

                    return lastSql;

                } else if ("mysql".equalsIgnoreCase(dbProductName)) {
                    // UPDATE test SET name=?, name2=? WHERE id=? AND id2=?
                    // 将SQL组装成 UPDATE kc27_copy1 a JOIN (SELECT '1' as KC27ID, '1' as BKC999, '1' as BKA001 UNION SELECT '2' as KC27ID, '2' as BKC999, '2' as BKA001) b USING(KC27ID,BKC999) SET a.BKA001=b.BKA001;

                    // 获取主键
                    List<String> keys = Arrays.asList(sql.substring(sql.indexOf("WHERE") + 6).replace("=?", "").split(" AND "));
                    // 添加修改的字段
                    List<String> fields = Arrays.asList(sql.substring(sql.indexOf("SET") + 4, sql.indexOf("WHERE") - 1).replace("=?", "").split(", "));

                    List<String> all = new ArrayList<>(keys);
                    all.addAll(fields);

                    StringBuilder sb = new StringBuilder();
                    sb.append("SELECT ");
                    for (String field : all) {
                        sb.append(" \001 AS ").append(field).append(",");
                    }

                    sb.deleteCharAt(sb.lastIndexOf(","));

                    // childSql: SELECT ? AS KC27ID,? AS BKC999,? AS BKA001
                    String childSql = sb.toString();

                    composeSql.append(sql, 0, sql.indexOf("SET")).append(" a JOIN (");

                    for (List<Map<String, ?>> value : values) {
                        composeSql.append(childSql);
                        for (String field : all) {
                            for (Map<String, ?> val : value) {
                                if (field.replace(prefix, "").replace(suffix, "").trim().equals(val.get("name").toString())) {
                                    int index = composeSql.indexOf("\001");
                                    // 替换 ?
                                    composeSql.deleteCharAt(index);
                                    composeSql.insert(index, formatValue((Integer) (val.get("type")), val.get("value")));
                                    break;
                                }
                            }
                        }
                        composeSql.append(" UNION ");
                    }

                    composeSql.delete(composeSql.lastIndexOf(" UNION "), composeSql.length());
                    composeSql.append(") b USING (");
                    for (String key : keys) {
                        composeSql.append(key).append(",");
                    }
                    composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                    composeSql.append(") SET ");
                    for (String field : fields) {
                        composeSql.append("a.").append(field).append("=b.").append(field).append(",");
                    }
                    composeSql.deleteCharAt(composeSql.lastIndexOf(","));

                    return composeSql.toString();

                } else {
                    // 拼接为多条SQL
                    for (List<Map<String, ?>> value : values) {
                        composeSql.append(sql.replace("?", "\001"));
                        for (Map<String, ?> val : value) {
                            int index = composeSql.indexOf("\001");
                            composeSql.deleteCharAt(index);
                            composeSql.insert(index, formatValue((Integer) (val.get("type")), val.get("value")));
                        }
                        composeSql.append(";");
                    }
                    return composeSql.toString();
                }

            } else if (sql.startsWith("DELETE")) {
                List<Map<String, ?>> data = values.get(0);
                // 主键数
                int keySize = data.size();
                // 单主键时将SQL构造成 delete from test where id in ('1','2')
                if (keySize == 1) {
                    composeSql.append(sql, 0, sql.indexOf("=")).append(" IN (");

                    for (List<Map<String, ?>> value : values) {
                        for (Map<String, ?> val : value) {
                            composeSql.append(formatValue((Integer) (val.get("type")), val.get("value"))).append(",");
                        }
                    }

                    composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                    composeSql.append(")");

                    return composeSql.toString();

                } else if (keySize > 1) {
                    // 联合主键

                    // 组装delete语句
                    if ("postgresql".equalsIgnoreCase(dbProductName)) {
                        // 将SQL构造成 delete from test using (values (1,'zhangsan'),(2,'lisi')) as tmp(id,name) where test.id=tmp.id and test.name=tmp.name;
                        composeSql.append(sql, 0, sql.indexOf("WHERE")).append(" t using (values ");

                        for (List<Map<String, ?>> value : values) {
                            composeSql.append("(");
                            for (Map<String, ?> val : value) {
                                composeSql.append(formatValue((Integer) (val.get("type")), val.get("value"))).append(",");
                            }
                            composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                            composeSql.append("),");
                        }

                        composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                        composeSql.append(") AS tmp(");
                        StringBuilder sb = new StringBuilder();
                        for (Map<String, ?> val : data) {
                            String name = prefix + val.get("name").toString() + suffix;
                            sb.append("t.").append(name).append("=tmp.").append(name).append(" AND ");
                            composeSql.append(name).append(",");
                        }
                        sb.delete(sb.lastIndexOf(" AND "), sb.length());
                        composeSql.deleteCharAt(composeSql.lastIndexOf(","));
                        composeSql.append(") WHERE ").append(sb);

                        String lastSql = composeSql.toString();
                        // 插入日期类型转换
                        for (Map<String, ?> val : data) {
                            lastSql = insertTypeCast(lastSql, prefix + val.get("name").toString() + suffix, (Integer) (val.get("type")));
                        }

                        return lastSql;
                    } else {
                        // 两种方案：1、以分号分隔拼成一条,，delete from test where id=1 and name='zz'
                        //          2、组合成一条，delete from test where (id=1 and name='zz') or (id=2 and name='yy')

//                        // 方案一
//                        for (List<Map<String, ?>> value : values) {
//                            composeSql.append(sql);
//                            for (Map<String, ?> val : value) {
//                                int index = composeSql.indexOf("?");
//                                composeSql.deleteCharAt(index);
//                                composeSql.insert(index, formatValue((Integer) (val.get("type")), val.get("value")));
//                            }
//                            composeSql.append(";");
//                        }
//                        logger.error("batch delete sql : {}", composeSql.toString());
//                        return composeSql.toString();

                        // 方案二
                        // 获取主键字段名
                        List<String> fields = new ArrayList<>();
                        for (Map<String, ?> val : data) {
                            fields.add(val.get("name").toString());
                        }

                        composeSql.append(sql, 0, sql.indexOf("WHERE") + 6);
                        for (List<Map<String, ?>> value : values) {
                            composeSql.append("(");

                            for (int i = 0, len = value.size(); i < len; i++) {
                                Map<String, ?> val = value.get(i);
                                composeSql.append(fields.get(i)).append("=").append(formatValue((Integer) (val.get("type")), val.get("value"))).append(" AND ");
                            }
                            composeSql.delete(composeSql.lastIndexOf(" AND "), composeSql.length());
                            composeSql.append(") OR ");
                        }
                        composeSql.delete(composeSql.lastIndexOf(" OR "), composeSql.length());
                        return composeSql.toString();
                    }
                }
            }
        }
        return "";
    }

    // postgresql sql 添加类型转换
    private String insertTypeCast(String sql, String fieldName, int type) {
        if (Types.DATE == type) {
            sql = sql.replace("tmp." + fieldName, "tmp." + fieldName + "::date");
        } else if (Types.TIME == type) {
            sql = sql.replace("tmp." + fieldName, "tmp." + fieldName + "::time");
        } else if (Types.TIMESTAMP == type) {
            sql = sql.replace("tmp." + fieldName, "tmp." + fieldName + "::timestamp");
        }
        return sql;
    }

    // 值转换
    private Object formatValue(int type, Object value) {
        if (null == value) {
            return "NULL";
        }
        if (Types.DATE == type) {
            if (value instanceof Long) {
                value = new SimpleDateFormat("yyyy-MM-dd").format(new Date((Long) value));
            } else if (value instanceof String) {
                value = new SimpleDateFormat("yyyy-MM-dd").format(Util.parseDate(value.toString()));
            }
        } else if (Types.TIME == type) {
            if (value instanceof Long) {
                value = new SimpleDateFormat("HH:mm:ss").format(new Date((Long) value));
            } else if (value instanceof String) {
                value = new SimpleDateFormat("HH:mm:ss").format(Util.parseDate(value.toString()));
            }
        } else if (Types.TIMESTAMP == type) {
            if (value instanceof Long) {
                value = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date((Long) value));
            } else if (value instanceof String) {
                value = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Util.parseDate(value.toString()));
            }
        } else if (Types.TINYINT == type || Types.SMALLINT == type || Types.INTEGER == type || Types.BIGINT == type || Types.DECIMAL == type ||
                Types.NUMERIC == type || Types.REAL == type || Types.FLOAT == type || Types.DOUBLE == type) {
            return value;
        }
        // 转义数据中的单引号
        return "'" + value.toString().replace("'", "''") + "'";
    }

    public String getDatabaseProductName() {
        try {
            return conn.getMetaData().getDatabaseProductName();
        } catch (SQLException throwables) {
            return "";
        }
    }

    public void commit() throws SQLException {
        getConn().commit();
        if (logger.isTraceEnabled()) {
            logger.trace("Batch executor commit " + idx.get() + " rows");
        }
        idx.set(0);
    }

    public void rollback() throws SQLException {
        getConn().rollback();
        if (logger.isTraceEnabled()) {
            logger.trace("Batch executor rollback " + idx.get() + " rows");
        }
        idx.set(0);
    }

    public void rollback(Savepoint savepoint) throws SQLException {
        getConn().rollback(savepoint);
        if (logger.isTraceEnabled()) {
            logger.trace("Batch executor rollback " + idx.get() + " rows");
        }
        idx.set(0);
    }

    @Override
    public void close() {
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                logger.error(e.getMessage(), e);
            } finally {
                conn = null;
            }
        }
    }
}
